# -*- coding:utf-8 -*-

import json
import requests
import platform
import hashlib
import uuid
import traceback
import asyncio
import asyncio.exceptions
import logging
import time
import urllib.error
import urllib.parse
import urllib.request
import socket
import websockets

from concurrent.futures import ThreadPoolExecutor
from Panda.Core.Log import logger as logger

VERSION_STRING = "0.18.1"
DINGTALK_OPENAPI_ENDPOINT = "https://api.dingtalk.com"


class Headers(object):
    CONTENT_TYPE_APPLICATION_JSON = "application/json"

    def __init__(self):
        self.app_id = None
        self.connection_id = None
        self.content_type = None
        self.message_id = None
        self.time = None
        self.topic = None
        self.extensions = {}
        # event fields start
        self.event_born_time = None
        self.event_corp_id = None
        self.event_id = None
        self.event_type = None
        self.event_unified_app_id = None
        # event fields end

    def __str__(self):
        fields = {
            "app_id": self.app_id,
            "connection_id": self.connection_id,
            "content_type": self.content_type,
            "message_id": self.message_id,
            "time": self.time,
            "topic": self.topic,
            "extensions": self.extensions,
        }
        if self.event_id is not None:
            fields["event_born_time"] = self.event_born_time
            fields["event_id"] = self.event_id
            fields["event_corp_id"] = self.event_corp_id
            fields["event_type"] = self.event_type
            fields["event_unified_app_id"] = self.event_unified_app_id
        return "Headers(%s)" % ", ".join(["%s=%s" % (k, v) for k, v in fields.items()])

    def to_dict(self):
        result = self.extensions.copy()
        if self.app_id is not None:
            result["appId"] = self.app_id
        if self.connection_id is not None:
            result["connectionId"] = self.connection_id
        if self.content_type is not None:
            result["contentType"] = self.content_type
        if self.message_id is not None:
            result["messageId"] = self.message_id
        if self.topic is not None:
            result["topic"] = self.topic
        if self.time is not None:
            result["time"] = str(self.time)
        return result

    @classmethod
    def from_dict(cls, d):
        headers = Headers()
        for name, value in d.items():
            if name == "appId":
                headers.app_id = value
            elif name == "connectionId":
                headers.connection_id = value
            elif name == "contentType":
                headers.content_type = value
            elif name == "messageId":
                headers.message_id = value
            elif name == "topic":
                headers.topic = value
            elif name == "time":
                headers.time = int(value)
            elif name == "eventBornTime":
                headers.event_born_time = int(value)
            elif name == "eventCorpId":
                headers.event_corp_id = value
            elif name == "eventId":
                headers.event_id = value
            elif name == "eventType":
                headers.event_type = value
            elif name == "eventUnifiedAppId":
                headers.event_unified_app_id = value
            else:
                headers.extensions[name] = value
        return headers


class EventMessage(object):
    TYPE = "EVENT"

    def __init__(self):
        self.spec_version = ""
        self.type = EventMessage.TYPE
        self.headers = Headers()
        self.data = {}
        self.extensions = {}

    def __str__(self):
        return (
            "EventMessage(spec_version=%s, type=%s, headers=%s, data=%s, extensions=%s)"
            % (self.spec_version, self.type, self.headers, self.data, self.extensions)
        )

    @classmethod
    def from_dict(cls, d):
        msg = EventMessage()
        data = ""
        for name, value in d.items():
            if name == "specVersion":
                msg.spec_version = value
            elif name == "data":
                data = value
            elif name == "type":
                pass
            elif name == "headers":
                msg.headers = Headers.from_dict(value)
            else:
                msg.extensions[name] = value
        if data:
            msg.data = json.loads(data)
        return msg


class CallbackMessage(object):
    TYPE = "CALLBACK"

    def __init__(self):
        self.spec_version = ""
        self.type = CallbackMessage.TYPE
        self.headers = Headers()
        self.data = {}
        self.extensions = {}

    def __str__(self):
        return (
            "CallbackMessage(spec_version=%s, type=%s, headers=%s, data=%s, extensions=%s)"
            % (self.spec_version, self.type, self.headers, self.data, self.extensions)
        )

    @classmethod
    def from_dict(cls, d):
        msg = CallbackMessage()
        data = ""
        for name, value in d.items():
            if name == "specVersion":
                msg.spec_version = value
            elif name == "data":
                data = value
            elif name == "type":
                pass
            elif name == "headers":
                msg.headers = Headers.from_dict(value)
            else:
                msg.extensions[name] = value
        if data:
            msg.data = json.loads(data)
        return msg


class SystemMessage(object):
    TYPE = "SYSTEM"
    TOPIC_DISCONNECT = "disconnect"

    def __init__(self):
        self.spec_version = ""
        self.type = SystemMessage.TYPE
        self.headers = Headers()
        self.data = {}
        self.extensions = {}

        @classmethod
        def from_dict(cls, d):
            msg = SystemMessage()
            data = ""
            for name, value in d.items():
                if name == "specVersion":
                    msg.spec_version = value
                elif name == "data":
                    data = value
                elif name == "type":
                    pass
                elif name == "headers":
                    msg.headers = Headers.from_dict(value)
                else:
                    msg.extensions[name] = value
            if data:
                msg.data = json.loads(data)
            return msg

    def __str__(self):
        return (
            "SystemMessage(spec_version=%s, type=%s, headers=%s, data=%s, extensions=%s)"
            % (
                self.spec_version,
                self.type,
                self.headers,
                self.data,
                self.extensions,
            )
        )

    @classmethod
    def from_dict(cls, d):
        msg = SystemMessage()
        data = ""
        for name, value in d.items():
            if name == "specVersion":
                msg.spec_version = value
            elif name == "data":
                data = value
            elif name == "type":
                pass
            elif name == "headers":
                msg.headers = Headers.from_dict(value)
            else:
                msg.extensions[name] = value
        if data:
            msg.data = json.loads(data)
        return msg


class AckMessage(object):
    STATUS_OK = 200
    STATUS_BAD_REQUEST = 400
    STATUS_NOT_IMPLEMENT = 404
    STATUS_SYSTEM_EXCEPTION = 500

    def __init__(self):
        self.code = AckMessage.STATUS_OK
        self.headers = Headers()
        self.message = ""
        self.data = {}

    def to_dict(self):
        return {
            "code": self.code,
            "headers": self.headers.to_dict(),
            "message": self.message,
            "data": json.dumps(self.data),
        }


class CallbackHandler(object):
    TOPIC_CARD_CALLBACK = "/v1.0/card/instances/callback"

    def __init__(self):
        self.dingtalk_client = None
        self.logger = logger

    def pre_start(self):
        return

    async def process(self, message: CallbackMessage):
        return AckMessage.STATUS_NOT_IMPLEMENT, "not implement"

    async def raw_process(self, callback_message: CallbackMessage):
        code, message = await self.process(callback_message)
        ack_message = AckMessage()
        ack_message.code = code
        ack_message.headers.message_id = callback_message.headers.message_id
        ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
        ack_message.data = {"response": message}
        return ack_message


class EventHandler(object):
    def __init__(self):
        self.dingtalk_client = None
        self.logger = logger

    def pre_start(self):
        return

    async def process(self, event: EventMessage):
        return AckMessage.STATUS_NOT_IMPLEMENT, "not implement"

    async def raw_process(self, event_message: EventMessage):
        code, message = await self.process(event_message)
        ack_message = AckMessage()
        ack_message.code = code
        ack_message.headers.message_id = event_message.headers.message_id
        ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
        ack_message.message = message
        ack_message.data = event_message.data
        return ack_message


class SystemHandler(object):
    def __init__(self):
        self.dingtalk_client = None
        self.logger = logger

    def pre_start(self):
        return

    async def process(self, message: SystemMessage):
        return AckMessage.STATUS_OK, "OK"

    async def raw_process(self, system_message: SystemMessage):
        code, message = await self.process(system_message)
        ack_message = AckMessage()
        ack_message.code = code
        ack_message.headers.message_id = system_message.headers.message_id
        ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
        ack_message.message = message
        ack_message.data = system_message.data
        return ack_message


class Credential(object):
    """初始化凭证"""

    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret


class DingTalkStreamClient(object):
    OPEN_CONNECTION_API = DINGTALK_OPENAPI_ENDPOINT + "/v1.0/gateway/connections/open"

    def __init__(self, credential: Credential, logger: logging.Logger = logger):
        self.credential: Credential = credential
        self.event_handler: EventHandler = EventHandler()
        self.callback_handler_map = {}
        self.system_handler: SystemHandler = SystemHandler()
        self.websocket = None  # create websocket client after connected
        self.logger = logger
        self._pre_started = False
        self._is_event_required = False
        self._access_token = {}

    def register_all_event_handler(self, handler: EventHandler):
        handler.dingtalk_client = self
        self.event_handler = handler
        self._is_event_required = True

    def register_callback_handler(self, topic, handler: CallbackHandler):
        handler.dingtalk_client = self
        self.callback_handler_map[topic] = handler

    def pre_start(self):
        if self._pre_started:
            return
        self._pre_started = True
        self.event_handler.pre_start()
        self.system_handler.pre_start()
        for handler in self.callback_handler_map.values():
            handler.pre_start()

    def open_connection(self):
        self.logger.info(
            "open connection, url=%s" % DingTalkStreamClient.OPEN_CONNECTION_API
        )
        request_headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": (
                "DingTalkStream/1.0 SDK/%s Python/%s "
                "(+https://github.com/open-dingtalk/dingtalk-stream-sdk-python)"
            )
            % (VERSION_STRING, platform.python_version()),
        }
        topics = []
        if self._is_event_required:
            topics.append({"type": "EVENT", "topic": "*"})
        for topic in self.callback_handler_map.keys():
            topics.append({"type": "CALLBACK", "topic": topic})
        request_body = json.dumps(
            {
                "clientId": self.credential.client_id,
                "clientSecret": self.credential.client_secret,
                "subscriptions": topics,
                "ua": "dingtalk-sdk-python/v%s" % VERSION_STRING,
                "localIp": self.get_host_ip(),
            }
        ).encode("utf-8")

        try:
            response = requests.post(
                DingTalkStreamClient.OPEN_CONNECTION_API,
                headers=request_headers,
                data=request_body,
            )
            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"open connection failed, error={e}, response.text={response.text}"
            )
            return None
        return response.json()

    async def start(self):
        self.pre_start()

        while True:
            connection = self.open_connection()

            if not connection:
                self.logger.error("open connection failed")
                time.sleep(10)
                continue
            self.logger.info("endpoint is %s", connection)

            uri = "%s?ticket=%s" % (
                connection["endpoint"],
                urllib.parse.quote_plus(connection["ticket"]),
            )
            self.logger.error(uri)
            async with websockets.connect(uri) as websocket:
                self.websocket = websocket
                async for raw_message in websocket:
                    json_message = json.loads(raw_message)
                    route_result = await self.route_message(json_message)
                    if route_result == "disconnect":
                        break
                # self.websocket.close()
        return

    def start_forever(self):
        while True:
            try:
                asyncio.run(self.start())
            except KeyboardInterrupt as e:
                break
            except (
                asyncio.exceptions.CancelledError,
                websockets.exceptions.ConnectionClosedError,
            ) as e:
                self.logger.error("network exception, error=%s", e)
                time.sleep(10)
                continue
            except Exception as e:
                time.sleep(3)
                self.logger.exception("unknown exception", e)
                continue

    async def route_message(self, json_message):
        result = ""
        msg_type = json_message.get("type", "")
        ack = None
        if msg_type == SystemMessage.TYPE:
            msg = SystemMessage.from_dict(json_message)
            ack = await self.system_handler.raw_process(msg)
            if msg.headers.topic == SystemMessage.TOPIC_DISCONNECT:
                result = "disconnect"
                self.logger.info(
                    "received disconnect topic=%s, message=%s",
                    msg.headers.topic,
                    json_message,
                )
            else:
                self.logger.warning(
                    "unknown message topic, topic=%s, message=%s",
                    msg.headers.topic,
                    json_message,
                )
        elif msg_type == EventMessage.TYPE:
            msg = EventMessage.from_dict(json_message)
            ack = await self.event_handler.raw_process(msg)
        elif msg_type == CallbackMessage.TYPE:
            msg = CallbackMessage.from_dict(json_message)
            handler = self.callback_handler_map.get(msg.headers.topic)
            if handler:
                ack = await handler.raw_process(msg)
            else:
                self.logger.warning(
                    "unknown callback message topic, topic=%s, message=%s",
                    msg.headers.topic,
                    json_message,
                )
        else:
            self.logger.warning("unknown message, content=%s", json_message)
        if ack:
            await self.websocket.send(json.dumps(ack.to_dict()))
        return result

    def get_host_ip(self):
        """查询本机ip"""
        ip = ""
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(("8.8.8.8", 80))
            ip = s.getsockname()[0]
        finally:
            s.close()
            return ip

    def reset_access_token(self):
        """reset token if open api return 401"""
        self._access_token = {}

    def get_access_token(self):
        now = int(time.time())
        if self._access_token and now < self._access_token["expireTime"]:
            return self._access_token["accessToken"]

        request_headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        values = {
            "appKey": self.credential.client_id,
            "appSecret": self.credential.client_secret,
        }
        try:
            response = requests.post(
                DINGTALK_OPENAPI_ENDPOINT + "/v1.0/oauth2/accessToken",
                headers=request_headers,
                data=json.dumps(values),
            )
            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"get dingtalk access token failed, error={e}, response.text={response.text}"
            )
            return None

        result = response.json()
        result["expireTime"] = (
            int(time.time()) + result["expireIn"] - (5 * 60)
        )  # reserve 5min buffer time
        self._access_token = result
        return self._access_token["accessToken"]

    # def upload_to_dingtalk(self, image_content, filetype='image', filename='image.png', mimetype='image/png'):
    #     """非必要函数，已经自行重写"""
    #     access_token = self.get_access_token()
    #     if not access_token:
    #         self.logger.error('upload_to_dingtalk failed, cannot get dingtalk access token')
    #         return None
    #     files = {
    #         'media': (filename, image_content, mimetype),
    #     }
    #     values = {
    #         'type': filetype,
    #     }
    #     upload_url = ('https://oapi.dingtalk.com/media/upload?access_token=%s'
    #                   ) % urllib.parse.quote_plus(access_token)
    #     try:
    #         response = requests.post(upload_url, data=values, files=files)
    #         if response.status_code == 401:
    #             self.reset_access_token()
    #         response.raise_for_status()
    #     except Exception as e:
    #         self.logger.error(f'upload to dingtalk failed, error={e}, response.text={response.text}')
    #         return None
    #     if 'media_id' not in response.json():
    #         self.logger.error('upload to dingtalk failed, error resonse is %s', response.json())
    #         raise Exception('upload failed, error=%s' % response.json())
    #     return response.json()['media_id']


class AtUser(object):
    def __init__(self):
        self.dingtalk_id = None
        self.staff_id = None
        self.extensions = {}

    @classmethod
    def from_dict(cls, d):
        user = AtUser()
        for name, value in d.items():
            if name == "dingtalkId":
                user.dingtalk_id = value
            elif name == "staffId":
                user.staff_id = value
            else:
                user.extensions[name] = value
        return user

    def to_dict(self):
        result = self.extensions.copy()
        if self.dingtalk_id is not None:
            result["dingtalkId"] = self.dingtalk_id
        if self.staff_id is not None:
            result["staffId"] = self.staff_id
        return result


class TextContent(object):
    content: str

    def __init__(self):
        self.content = None
        self.extensions = {}

    def __str__(self):
        return "TextContent(content=%s)" % self.content

    @classmethod
    def from_dict(cls, d):
        content = TextContent()
        for name, value in d.items():
            if name == "content":
                content.content = value
            else:
                content.extensions[name] = value
        return content

    def to_dict(self):
        result = self.extensions.copy()
        if self.content is not None:
            result["content"] = self.content
        return result


class ImageContent(object):

    def __init__(self):
        self.download_code = None

    @classmethod
    def from_dict(cls, d):
        content = ImageContent()
        for name, value in d.items():
            if name == "downloadCode":
                content.download_code = value
        return content

    def to_dict(self):
        result = {}
        if self.download_code is not None:
            result["downloadCode"] = self.download_code
        return result


class RichTextContent(object):

    def __init__(self):
        self.rich_text_list = None

    @classmethod
    def from_dict(cls, d):
        content = RichTextContent()
        content.rich_text_list = []
        for name, value in d.items():
            if name == "richText":
                content.rich_text_list = value
        return content

    def to_dict(self):
        result = {}
        if self.rich_text_list is not None:
            result["richText"] = self.rich_text_list
        return result


class HostingContext(object):
    """
    托管人的上下文
    """

    def __init__(self):
        self.user_id = ""
        self.nick = ""

    def to_dict(self):
        result = {
            "userId": self.user_id,
            "nick": self.nick,
        }
        return result


class ConversationMessage(object):
    """
    历史消息状态
    """

    def __init__(self):
        self.read_status = ""
        self.sender_user_id = ""
        self.send_time = 0

    def read_by_me(self) -> bool:
        """
        消息是否被我已读
        :return:
        """
        return self.read_status == "2"

    def to_dict(self):
        result = {
            "readStatus": self.read_status,
            "senderUserId": self.sender_user_id,
            "sendTime": self.send_time,
        }
        return result


class ChatbotMessage(object):
    TOPIC = "/v1.0/im/bot/messages/get"
    DELEGATE_TOPIC = "/v1.0/im/bot/messages/delegate"
    text: TextContent

    def __init__(self):
        self.is_in_at_list = None
        self.session_webhook = None
        self.sender_nick = None
        self.robot_code = None
        self.session_webhook_expired_time = None
        self.message_id = None
        self.sender_id = None
        self.chatbot_user_id = None
        self.conversation_id = None
        self.is_admin = None
        self.create_at = None
        self.text = None
        self.conversation_type = None
        self.at_users = []
        self.chatbot_corp_id = None
        self.sender_corp_id = None
        self.conversation_title = None
        self.message_type = None
        self.image_content = None
        self.rich_text_content = None
        self.sender_staff_id = None
        self.hosting_context: HostingContext = None
        self.conversation_msg_context = None

        self.extensions = {}

    @classmethod
    def from_dict(cls, d):
        msg = ChatbotMessage()
        data = ""
        for name, value in d.items():
            if name == "isInAtList":
                msg.is_in_at_list = value
            elif name == "sessionWebhook":
                msg.session_webhook = value
            elif name == "senderNick":
                msg.sender_nick = value
            elif name == "robotCode":
                msg.robot_code = value
            elif name == "sessionWebhookExpiredTime":
                msg.session_webhook_expired_time = int(value)
            elif name == "msgId":
                msg.message_id = value
            elif name == "senderId":
                msg.sender_id = value
            elif name == "chatbotUserId":
                msg.chatbot_user_id = value
            elif name == "conversationId":
                msg.conversation_id = value
            elif name == "isAdmin":
                msg.is_admin = value
            elif name == "createAt":
                msg.create_at = value
            elif name == "conversationType":
                msg.conversation_type = value
            elif name == "atUsers":
                msg.at_users = [AtUser.from_dict(i) for i in value]
            elif name == "chatbotCorpId":
                msg.chatbot_corp_id = value
            elif name == "senderCorpId":
                msg.sender_corp_id = value
            elif name == "conversationTitle":
                msg.conversation_title = value
            elif name == "msgtype":
                msg.message_type = value
                if value == "text":
                    msg.text = TextContent.from_dict(d["text"])
                elif value == "picture":
                    msg.image_content = ImageContent.from_dict(d["content"])
                elif value == "richText":
                    msg.rich_text_content = RichTextContent.from_dict(d["content"])
            elif name == "senderStaffId":
                msg.sender_staff_id = value
            elif name == "hostingContext":
                msg.hosting_context = HostingContext()
                msg.hosting_context.user_id = value["userId"]
                msg.hosting_context.nick = value["nick"]
            elif name == "conversationMsgContext":
                msg.conversation_msg_context = []
                for v in value:
                    conversation_msg = ConversationMessage()
                    conversation_msg.read_status = v["readStatus"]
                    conversation_msg.send_time = v["sendTime"]
                    conversation_msg.sender_user_id = v["senderUserId"]

                    msg.conversation_msg_context.append(conversation_msg)
            else:
                msg.extensions[name] = value
        return msg

    def to_dict(self):
        result = self.extensions.copy()
        if self.is_in_at_list is not None:
            result["isInAtList"] = self.is_in_at_list
        if self.session_webhook is not None:
            result["sessionWebhook"] = self.session_webhook
        if self.sender_nick is not None:
            result["senderNick"] = self.sender_nick
        if self.robot_code is not None:
            result["robotCode"] = self.robot_code
        if self.session_webhook_expired_time is not None:
            result["sessionWebhookExpiredTime"] = self.session_webhook_expired_time
        if self.message_id is not None:
            result["msgId"] = self.message_id
        if self.sender_id is not None:
            result["senderId"] = self.sender_id
        if self.chatbot_user_id is not None:
            result["chatbotUserId"] = self.chatbot_user_id
        if self.conversation_id is not None:
            result["conversationId"] = self.conversation_id
        if self.is_admin is not None:
            result["isAdmin"] = self.is_admin
        if self.create_at is not None:
            result["createAt"] = self.create_at
        if self.text is not None:
            result["text"] = self.text.to_dict()
        if self.image_content is not None:
            result["content"] = self.image_content.to_dict()
        if self.rich_text_content is not None:
            result["content"] = self.rich_text_content.to_dict()
        if self.conversation_type is not None:
            result["conversationType"] = self.conversation_type
        if self.at_users is not None:
            result["atUsers"] = [i.to_dict() for i in self.at_users]
        if self.chatbot_corp_id is not None:
            result["chatbotCorpId"] = self.chatbot_corp_id
        if self.sender_corp_id is not None:
            result["senderCorpId"] = self.sender_corp_id
        if self.conversation_title is not None:
            result["conversationTitle"] = self.conversation_title
        if self.message_type is not None:
            result["msgtype"] = self.message_type
        if self.sender_staff_id is not None:
            result["senderStaffId"] = self.sender_staff_id
        if self.hosting_context is not None:
            result["hostingContext"] = self.hosting_context.to_dict()
        if self.conversation_msg_context is not None:
            result["conversationMsgContext"] = [
                v.to_dict() for v in self.conversation_msg_context
            ]
        return result

    def get_text_list(self):
        if self.message_type == "text":
            return [self.text.content]
        elif self.message_type == "richText":
            text = []
            for item in self.rich_text_content.rich_text_list:
                if "text" in item:
                    text.append(item["text"])

            return text

    def get_image_list(self):
        if self.message_type == "picture":
            return [self.image_content.download_code]
        elif self.message_type == "richText":
            images = []
            for item in self.rich_text_content.rich_text_list:
                if "downloadCode" in item:
                    images.append(item["downloadCode"])

            return images

    def __str__(self):
        return (
            "ChatbotMessage(message_type=%s, text=%s, sender_nick=%s, conversation_title=%s)"
            % (
                self.message_type,
                self.text,
                self.sender_nick,
                self.conversation_title,
            )
        )


class ChatbotHandler(CallbackHandler):

    def __init__(self):
        super(ChatbotHandler, self).__init__()

    def extract_text_from_incoming_message(
        self, incoming_message: ChatbotMessage
    ) -> list:
        """
        获取文本列表
        :param incoming_message:
        :return: text list。如果是纯文本消息，结果列表中只有一个元素；如果是富文本消息，结果是长列表，按富文本消息的逻辑分割，大致是按换行符分割的。
        """
        return incoming_message.get_text_list()

    def extract_image_from_incoming_message(
        self, incoming_message: ChatbotMessage
    ) -> list:
        """
        获取用户发送的图片，重新上传，获取新的mediaId列表
        :param incoming_message:
        :return: mediaid list
        """
        image_list = incoming_message.get_image_list()
        if image_list is None or len(image_list) == 0:
            return None

        mediaids = []
        for download_code in image_list:
            download_url = self.get_image_download_url(download_code)

            image_content = requests.get(download_url)
            mediaid = self.dingtalk_client.upload_to_dingtalk(
                image_content.content,
                filetype="image",
                filename="image.png",
                mimetype="image/png",
            )

            mediaids.append(mediaid)

        return mediaids

    def get_image_download_url(self, download_code: str) -> str:
        """
        根据downloadCode获取下载链接 https://open.dingtalk.com/document/isvapp/download-the-file-content-of-the-robot-receiving-message
        :param download_code:
        :return:
        """
        access_token = self.dingtalk_client.get_access_token()
        if not access_token:
            self.logger.error(
                "send_off_duty_prompt failed, cannot get dingtalk access token"
            )
            return None

        request_headers = {
            "Content-Type": "application/json",
            "Accept": "*/*",
            "x-acs-dingtalk-access-token": access_token,
            "User-Agent": (
                "DingTalkStream/1.0 SDK/0.1.0 Python/%s "
                "(+https://github.com/open-dingtalk/dingtalk-stream-sdk-python)"
            )
            % platform.python_version(),
        }

        values = {
            "robotCode": self.dingtalk_client.credential.client_id,
            "downloadCode": download_code,
        }

        url = DINGTALK_OPENAPI_ENDPOINT + "/v1.0/robot/messageFiles/download"

        try:
            response = requests.post(
                url, headers=request_headers, data=json.dumps(values)
            )

            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"get_image_download_url, error={e}, response.text={response.text}"
            )
            return ""
        return response.json()["downloadUrl"]

    def set_off_duty_prompt(self, text: str, title: str = "", logo: str = ""):
        """
        设置离线提示词，需要使用OpenAPI，当前仅支持自建应用。
        :param text: 离线提示词，支持markdown
        :param title: 机器人名称，默认："钉钉Stream机器人"
        :param logo: 机器人logo，默认："@lALPDfJ6V_FPDmvNAfTNAfQ"
        :return:
        """
        access_token = self.dingtalk_client.get_access_token()
        if not access_token:
            self.logger.error(
                "send_off_duty_prompt failed, cannot get dingtalk access token"
            )
            return None

        if title is None or title == "":
            title = "钉钉Stream机器人"

        if logo is None or logo == "":
            logo = "@lALPDfJ6V_FPDmvNAfTNAfQ"

        prompt_card_data = {
            "title": "离线",
            "text": "机器人已经离线！",
        }

        request_headers = {
            "Content-Type": "application/json",
            "Accept": "*/*",
            "x-acs-dingtalk-access-token": access_token,
            "User-Agent": (
                "DingTalkStream/1.0 SDK/0.1.0 Python/%s "
                "(+https://github.com/open-dingtalk/dingtalk-stream-sdk-python)"
            )
            % platform.python_version(),
        }

        values = {
            "robotCode": self.dingtalk_client.credential.client_id,
            "cardData": json.dumps(prompt_card_data),
            "cardTemplateId": "StandardCard",
        }

        url = (
            DINGTALK_OPENAPI_ENDPOINT
            + "/v1.0/innerApi/robot/stream/away/template/update"
        )

        try:
            response = requests.post(
                url, headers=request_headers, data=json.dumps(values)
            )

            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"set_off_duty_prompt, error={e}, response.text={response.text}"
            )
            return response.status_code
        return response.json()

    def reply_text(self, text: str, incoming_message: ChatbotMessage):
        request_headers = {
            "Content-Type": "application/json",
            "Accept": "*/*",
        }
        values = {
            "msgtype": "text",
            "text": {
                "content": text,
            },
            "at": {
                "atUserIds": [incoming_message.sender_staff_id],
            },
        }
        try:
            response = requests.post(
                incoming_message.session_webhook,
                headers=request_headers,
                data=json.dumps(values),
            )
            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"reply text failed, error={e}, response.text={response.text}"
            )
            return None
        return response.json()

    def reply_markdown(self, title: str, text: str, incoming_message: ChatbotMessage):
        request_headers = {
            "Content-Type": "application/json",
            "Accept": "*/*",
        }
        values = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text,
            },
            "at": {
                "atUserIds": [incoming_message.sender_staff_id],
            },
        }
        try:
            response = requests.post(
                incoming_message.session_webhook,
                headers=request_headers,
                data=json.dumps(values),
            )
            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"reply markdown failed, error={e}, response.text={response.text}"
            )
            return None
        return response.json()

    def reply_card(
        self,
        card_data: dict,
        incoming_message: ChatbotMessage,
        at_sender: bool = False,
        at_all: bool = False,
        **kwargs,
    ) -> str:
        """
        机器人回复互动卡片。由于 sessionWebhook 不支持发送互动卡片，所以需要使用 OpenAPI，当前仅支持自建应用。
        https://open.dingtalk.com/document/orgapp/robots-send-interactive-cards
        :param card_data: 卡片数据内容，interactive_card.py 中有一些简单的样例，高阶需求请至卡片搭建平台：https://card.dingtalk.com/card-builder
        :param incoming_message: 回调数据源
        :param at_sender: 是否at发送人
        :param at_all: 是否at所有人
        :param kwargs: 其他参数，具体可参考文档。
        :return:
        """
        access_token = self.dingtalk_client.get_access_token()
        if not access_token:
            self.logger.error(
                "simple_reply_interactive_card_only_for_inner_app failed, cannot get dingtalk access token"
            )
            return None

        request_headers = {
            "Content-Type": "application/json",
            "Accept": "*/*",
            "x-acs-dingtalk-access-token": access_token,
            "User-Agent": (
                "DingTalkStream/1.0 SDK/0.1.0 Python/%s "
                "(+https://github.com/open-dingtalk/dingtalk-stream-sdk-python)"
            )
            % platform.python_version(),
        }

        card_biz_id = self._gen_card_id(incoming_message)
        body = {
            "cardTemplateId": "StandardCard",
            "robotCode": self.dingtalk_client.credential.client_id,
            "cardData": json.dumps(card_data),
            "sendOptions": {
                # "atUserListJson": "String",
                # "atAll": at_all,
                # "receiverListJson": "String",
                # "cardPropertyJson": "String"
            },
            "cardBizId": card_biz_id,
        }

        if incoming_message.conversation_type == "2":
            body["openConversationId"] = incoming_message.conversation_id
        elif incoming_message.conversation_type == "1":
            single_chat_receiver = {"userId": incoming_message.sender_staff_id}
            body["singleChatReceiver"] = json.dumps(single_chat_receiver)

        if at_all:
            body["sendOptions"]["atAll"] = True
        else:
            body["sendOptions"]["atAll"] = False

        if at_sender:
            user_list_json = [
                {
                    "nickName": incoming_message.sender_nick,
                    "userId": incoming_message.sender_staff_id,
                }
            ]
            body["sendOptions"]["atUserListJson"] = json.dumps(
                user_list_json, ensure_ascii=False
            )
        body.update(**kwargs)

        url = DINGTALK_OPENAPI_ENDPOINT + "/v1.0/im/v1.0/robot/interactiveCards/send"
        try:
            response = requests.post(url, headers=request_headers, json=body)
            response.raise_for_status()

            return card_biz_id
        except Exception as e:
            self.logger.error(
                f"reply card failed, error={e}, response.text={response.text}"
            )
            return ""

    def update_card(self, card_biz_id: str, card_data: dict):
        """
        更新机器人发送互动卡片（普通版）。
        https://open.dingtalk.com/document/orgapp/update-the-robot-to-send-interactive-cards
        :param card_biz_id: 唯一标识一张卡片的外部ID（卡片幂等ID，可用于更新或重复发送同一卡片到多个群会话）。需与 self.reply_card 接口返回的 card_biz_id 保持一致。
        :param card_data: 要更新的卡片数据内容。详情参考卡片搭建平台：https://card.dingtalk.com/card-builder
        :return:
        """
        access_token = self.dingtalk_client.get_access_token()
        if not access_token:
            self.logger.error("update_card failed, cannot get dingtalk access token")
            return None

        request_headers = {
            "Content-Type": "application/json",
            "Accept": "*/*",
            "x-acs-dingtalk-access-token": access_token,
            "User-Agent": (
                "DingTalkStream/1.0 SDK/0.1.0 Python/%s "
                "(+https://github.com/open-dingtalk/dingtalk-stream-sdk-python)"
            )
            % platform.python_version(),
        }

        values = {
            "cardBizId": card_biz_id,
            "cardData": json.dumps(card_data),
        }
        url = DINGTALK_OPENAPI_ENDPOINT + "/v1.0/im/robots/interactiveCards"
        try:
            response = requests.put(
                url, headers=request_headers, data=json.dumps(values)
            )
            response.raise_for_status()
        except Exception as e:
            self.logger.error(
                f"update card failed, error={e}, response.text={response.text}"
            )
            return response.status_code
        return response.json()

    @staticmethod
    def _gen_card_id(msg: ChatbotMessage):
        factor = "%s_%s_%s_%s_%s" % (
            msg.sender_id,
            msg.sender_corp_id,
            msg.conversation_id,
            msg.message_id,
            str(uuid.uuid1()),
        )
        m = hashlib.sha256()
        m.update(factor.encode("utf-8"))
        return m.hexdigest()


class AsyncChatbotHandler(ChatbotHandler):
    """
    多任务执行handler，注意：process函数重载的时候不要用 async
    """

    async_executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=8)

    def __init__(self, max_workers: int = 8):
        super(AsyncChatbotHandler, self).__init__()
        self.async_executor = ThreadPoolExecutor(max_workers=max_workers)

    def process(self, message):
        """
        不要用 async 修饰
        :param message:
        :return:
        """
        return AckMessage.STATUS_NOT_IMPLEMENT, "not implement"

    async def raw_process(self, callback_message: CallbackMessage):
        def func():
            try:
                self.process(callback_message)
            except Exception as e:
                self.logger.error(traceback.format_exc())

        self.async_executor.submit(func)

        ack_message = AckMessage()
        ack_message.code = AckMessage.STATUS_OK
        ack_message.headers.message_id = callback_message.headers.message_id
        ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
        ack_message.message = "ok"
        ack_message.data = callback_message.data
        return ack_message
